package com.itcast.spark.baseTree

import org.apache.spark.ml.classification.DecisionTreeClassifier
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature._
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.ml.tuning.{CrossValidator, CrossValidatorModel, ParamGridBuilder, TrainValidationSplit, TrainValidationSplitModel}
import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * DESC:这个代码是模板代码，主要的作用是通过构建机器学习全流程实现模型构建和预测
 * 1-准备环境
 * 2-准备数据源
 * 3-数据的基本信息的查看
 * 4-特征工程
 * 5-准备算法
 * 6-模型的超参数的校验
 * 7-模型训练
 * 8-模型预测
 * 9-模型保存
 * 10-已经保存的模型实现预测
 */
object _02IrisMllibModelTrainTestSplit {
  def main(args: Array[String]): Unit = {
    //1-准备环境
    val conf: SparkConf = new SparkConf().setAppName("SparkLibSvmDtcModel").setMaster("local[*]")
    val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    //2-准备数据源
    val dataDF: DataFrame = spark.read.format("csv").option("header", true).option("inferschema", true).load("./datasets/mldata/iris.csv")
    //3-数据的基本信息的查看
    //dataDF.printSchema()
    //dataDF.show(false)
    val split: Array[Dataset[Row]] = dataDF.randomSplit(Array(0.8, 0.2), seed = 1234L)
    val trainingSet: Dataset[Row] = split(0)
    val testSet: Dataset[Row] = split(1)
    //4-特征工程
    val stringIndexer: StringIndexer = new StringIndexer().setInputCol("class").setOutputCol("classlabel")
    val indexerModel: StringIndexerModel = stringIndexer.fit(trainingSet)
    val vectorAssembler: VectorAssembler = new VectorAssembler().setInputCols(Array("sepal_length", "sepal_width", "petal_length", "petal_width")).setOutputCol("features")
    val standardScaler: StandardScaler = new StandardScaler().setInputCol("features").setOutputCol("Stadfeatures")
    //5-准备算法
    //7-模型训练
    val decisionTreeClassifier: DecisionTreeClassifier = new DecisionTreeClassifier()
      .setFeaturesCol("Stadfeatures")
      .setLabelCol("classlabel")
      .setProbabilityCol("probability")
      .setPredictionCol("prediction") //预测为0-1的数值，后续可以对应到底是那个花
      .setMaxDepth(5)
      .setImpurity("gini")
    val indexToString: IndexToString = new IndexToString().setInputCol("prediction").setOutputCol("beforeIndex").setLabels(indexerModel.labels)
    val pipeline: Pipeline = new Pipeline().setStages(Array(stringIndexer, vectorAssembler, standardScaler, decisionTreeClassifier, indexToString))
    //val pipelineModel: PipelineModel = pipeline.fit(trainingSet)
    //8-模型预测
    //val y_pred_train: DataFrame = pipelineModel.transform(trainingSet)
    //val y_pred_test: DataFrame = pipelineModel.transform(testSet)
    //y_pred_train.show()
    //9-模型校验
    val evaluator: MulticlassClassificationEvaluator = new MulticlassClassificationEvaluator().setLabelCol("classlabel").setMetricName("accuracy").setPredictionCol("prediction")
    //6-模型的超参数的校验
    val builder: Array[ParamMap] = new ParamGridBuilder()
      .addGrid(decisionTreeClassifier.impurity, Array("gini", "entropy"))
      .addGrid(decisionTreeClassifier.maxDepth, Array(4, 5, 6))
      .build()
    /*    val validator: CrossValidator = new CrossValidator()
          .setEstimator(pipeline) //model
          .setEstimatorParamMaps(builder) //设定那些参数
          .setEvaluator(evaluator) //设置校验方式
          .setNumFolds(3)*/
    val validator: TrainValidationSplit = new TrainValidationSplit()
      .setEstimator(pipeline) //model
      .setEstimatorParamMaps(builder) //设定那些参数
      .setEvaluator(evaluator) //设置校验方式
      .setTrainRatio(0.8)
    val crossModel: TrainValidationSplitModel = validator.fit(trainingSet)
    //8-模型预测
    val y_pred_train: DataFrame = crossModel.transform(trainingSet)
    val y_pred_test: DataFrame = crossModel.transform(testSet)
    //y_pred_train.show()
    val accuracy_train: Double = evaluator.evaluate(y_pred_train)
    val accuracy_test: Double = evaluator.evaluate(y_pred_test)
    println("Accuracy train result is:", accuracy_train)
    println("Accuracy test result is:", accuracy_test)
    //(Accuracy train result is:,0.9920634920634921)
    //(Accuracy test result is:,0.9583333333333334)
    val model: PipelineModel = crossModel.bestModel.asInstanceOf[PipelineModel]
    println(model.stages(3).extractParamMap())
    /*{
      dtc_47ed21d60c9d-cacheNodeIds: false,
      dtc_47ed21d60c9d-checkpointInterval: 10,
      dtc_47ed21d60c9d-featuresCol: Stadfeatures,
      dtc_47ed21d60c9d-impurity: gini,
      dtc_47ed21d60c9d-labelCol: classlabel,
      dtc_47ed21d60c9d-maxBins: 32,
      dtc_47ed21d60c9d-maxDepth: 4,
      dtc_47ed21d60c9d-maxMemoryInMB: 256,
      dtc_47ed21d60c9d-minInfoGain: 0.0,
      dtc_47ed21d60c9d-minInstancesPerNode: 1,
      dtc_47ed21d60c9d-predictionCol: prediction,
      dtc_47ed21d60c9d-probabilityCol: probability,
      dtc_47ed21d60c9d-rawPredictionCol: rawPrediction,
      dtc_47ed21d60c9d-seed: 159147643
    }*/
    //9-模型保存
    //crossModel.save("")
    //10-已经保存的模型实现预测
  }
}
